package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class Dwd_step_back {

    public static void main(String[] args) {
        // 1.创建TableEnv
        TableEnvironment tableEnv = getTableEnv();

        // 2.输入表 读取数据
        createInputTable(tableEnv);

        // 3.数据处理
//        Table resultTable = handle(tableEnv);

        // 4.映射表 写入Kafka表中
//        createOutputTable(tableEnv);

        // 5.保存数据
//        saveToSink(tableEnv, resultTable);
    }

    private static void createInputTable(TableEnvironment tableEnv) {

    }

    private static TableEnvironment getTableEnv() {
        //1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);

        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tabEnv;
    }
}
